package com.coolpad.job.device.dao;

import com.coolpad.job.device.model.GeoFullInfoPO;
import com.coolpad.job.device.util.HBaseConvertUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.BufferedMutator;
import org.apache.hadoop.hbase.client.BufferedMutatorParams;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.IntStream;

@Slf4j
//@Component
public class HBaseDao {



    private List<Connection> connectionPool = new CopyOnWriteArrayList<>();
    private List<BufferedMutator> bufferedPool = new CopyOnWriteArrayList<>();

    private final int connectionNum = 3;

    private int connectIndex = 0;

    @Autowired
    private Configuration configuration;

    @Value("${hbase.table.name.geo.fill.info}")
    private String gfiTableName;

    private final BufferedMutator.ExceptionListener listener = (e, mutator) -> {
        for (int i = 0; i < e.getNumExceptions(); i++) {
            log.error("Failed to sent put {}",e.getRow(i));
        }
    };

    @Autowired
    public void init(Configuration configuration, @Value("${hbase.table.name.geo.fill.info}") String gfiTableName) {
        this.configuration = configuration;
        this.gfiTableName = gfiTableName;
        initPool();
        new Thread(() -> {
            while (true){
                try {
                    Thread.sleep(30*1000);
                } catch (InterruptedException e) {
                    log.error("",e);
                }
                checkConnectionClose();
            }
        }){{super.setName("HBase-connect-keepAlive");}}.start();
    }

    private void initPool() {
        IntStream.range(0,connectionNum)
                .forEach(num -> {
                    try {
                        Connection connection = createConnection(configuration);
                        BufferedMutator bufferedMutator = createBufferedMutator(connection);
                        connectionPool.add(connection);
                        bufferedPool.add(bufferedMutator);
                    } catch (IOException e) {
                        log.error("", e);
                    }
                });
    }

    private void checkConnectionClose(){
        IntStream.range(0,connectionNum)
                .forEach(n ->{
                    Connection connect = connectionPool.get(n);
                    if(connect.isClosed()){
                        try {
                            Connection connection = createConnection(configuration);
                            BufferedMutator bufferedMutator = createBufferedMutator(connection);
                            connectionPool.set(n,connection);
                            bufferedPool.set(n,bufferedMutator);
                        } catch (IOException e) {
                            log.error("",e);
                        }
                    }
                });
    }

    public Connection getConnection(){
        return connectionPool.get(connectIndex = ++connectIndex % connectionNum);
    }

    public BufferedMutator getBufferedMutator(){
        return bufferedPool.get(connectIndex = ++connectIndex % connectionNum);
    }

    public Connection createConnection(org.apache.hadoop.conf.Configuration configuration) throws IOException {
        return ConnectionFactory.createConnection(configuration);
    }

    /**
     *
     * @return
     * @throws IOException
     */
    public BufferedMutator createBufferedMutator(Connection connection) throws IOException {
        checkConnectionClose();
        BufferedMutator mutator;
        TableName tName = TableName.valueOf(gfiTableName);
        BufferedMutatorParams params = new BufferedMutatorParams(tName).listener(listener);
        params.writeBufferSize(5*1024*1024); // 可以自己设定阈值 5M 达到5M则提交一次
        mutator = connection.getBufferedMutator(params);
        return mutator;
    }

    /**
     * 自动提交
     * @param tableName
     * @param putList
     * @return
     */
    public boolean autoPut(String tableName, List<Put> putList) {
        Table table = null;
        try {
            TableName tName = TableName.valueOf(tableName);
            table = getConnection().getTable(tName);
            //数据量达到某个阈值时提交，不达到不提交
            table.put(putList);
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        } finally {
            try {
                //提交一次
                if (table != null) table.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 手动提交
     * @param putList
     * @return
     */
    public boolean put(String tableName, List<Put> putList) {
        BufferedMutator mutator = null;
        TableName tName = TableName.valueOf(tableName);
        BufferedMutatorParams params = new BufferedMutatorParams(tName).listener(listener);
        // 可以自己设定阈值 5M 达到5M则提交一次
        params.writeBufferSize(5*1024*1024);
        try {
            mutator = getConnection().getBufferedMutator(params);
            // 数据量达到5M时会自动提交一次
            mutator.mutate(putList);
//            mutator.flush(); // 手动提交一次
        } catch(Exception e) {
            log.error("",e);
            return false;
        } finally {
            try {
                // 提交一次
                if(mutator != null)mutator.close();
            }
            catch(Exception e) {
                log.error("",e);
            }
        }
        return true;
    }

    public boolean put(List<GeoFullInfoPO> pos) {
        List<Put> puts = HBaseConvertUtils.toPuts(pos);
        try {
            getBufferedMutator().mutate(puts);
        } catch (Exception e) {
            log.error("",e);
            return false;
        }
        return true;
    }

    public void flush(){
        bufferedPool.forEach(bufferedMutator -> {
            try {
                bufferedMutator.flush();
            } catch (IOException e) {
                log.error("flush buffer exception",e);
            }
        });
    }
}
